-module (osmos_table_format).

-export ([ binary_replace/1,
	   term_replace/1,
	   term_sum_uint63_delete/1,
	   binary_sum_uint64_vector/1 ]).

-ifdef (HAVE_EUNIT).
-include_lib ("eunit/include/eunit.hrl").
-endif.

-include ("osmos.hrl").
-include ("osmos_format.hrl").

%
% public
%

binary_replace (BlockSize) when is_integer (BlockSize),
				BlockSize >= ?OSMOS_MIN_BLOCK_SIZE ->
  BinaryFormat = osmos_format:binary (),
  BinaryCompare = osmos_compare:binary (),
  Merge = fun merge_return_later/3,
  ShortCircuitAlways = fun short_circuit_always/2,
  DeleteNever = fun delete_never/2,
  #osmos_table_format { block_size = BlockSize,
			 key_format = BinaryFormat,
			 key_compare = BinaryCompare,
			 value_format = BinaryFormat,
			 merge = Merge,
			 short_circuit = ShortCircuitAlways,
			 delete = DeleteNever }.

term_replace (BlockSize) when is_integer (BlockSize),
			      BlockSize >= ?OSMOS_MIN_BLOCK_SIZE ->
  TermFormat = osmos_format:term (),
  TermCompare = osmos_compare:term (),
  Merge = fun merge_return_later/3,
  ShortCircuitAlways = fun short_circuit_always/2,
  DeleteNever = fun delete_never/2,
  #osmos_table_format { block_size = BlockSize,
			 key_format = TermFormat,
			 key_compare = TermCompare,
			 value_format = TermFormat,
			 merge = Merge,
			 short_circuit = ShortCircuitAlways,
			 delete = DeleteNever }.

term_sum_uint63_delete (BlockSize)
	when is_integer (BlockSize), BlockSize >= ?OSMOS_MIN_BLOCK_SIZE ->
  KeyFormat = osmos_format:term (),
  KeyCompare = osmos_compare:term (),
  ValueToBinary = fun
		    (N) when is_integer (N), N >= 0, N < 16#7000000000000000 ->
		      <<0:1, N:63>>;
		    (delete) ->
		      <<>>
		  end,
  ValueFromBinary = fun
		      (<<_:1, N:63>>) -> N
		    end,
  ValueFormat = #osmos_format { to_binary = ValueToBinary,
				 from_binary = ValueFromBinary },
  Merge = fun
	    (_K, _Old, <<>>) -> <<>>;
	    (_K, <<>>, <<_:1, N:63>>) -> <<1:1, N:63>>;
	    (_K, _Old, New = <<1:1, _:63>>) -> New;
	    (_K, <<R:1, A:63>>, <<0:1, B:63>>) -> S = A + B, <<R:1, S:63>>
	  end,
  ShortCircuit = fun
		   (_K, <<>>) -> true;
		   (_K, <<1:1, _:63>>) -> true;
		   (_K, <<0:1, _:63>>) -> false
		 end,
  DeleteEmpty = fun delete_empty/2,
  #osmos_table_format { block_size = BlockSize,
			 key_format = KeyFormat,
			 key_compare = KeyCompare,
			 value_format = ValueFormat,
			 merge = Merge,
			 short_circuit = ShortCircuit,
			 delete = DeleteEmpty }.

binary_sum_uint64_vector (BlockSize)
	when is_integer (BlockSize), BlockSize > ?OSMOS_MIN_BLOCK_SIZE ->
  KeyFormat = osmos_format:binary (),
  KeyCompare = osmos_compare:binary (),
  ValueFormat = osmos_format:uint64_vector (),
  MergeSum = fun merge_uint64_vector_sum/3,
  ShortCircuitNever = fun short_circuit_never/2,
  DeleteNever = fun delete_never/2,
  #osmos_table_format { block_size = BlockSize,
			 key_format = KeyFormat,
			 key_compare = KeyCompare,
			 value_format = ValueFormat,
			 merge = MergeSum,
			 short_circuit = ShortCircuitNever,
			 delete = DeleteNever }.

%
% private
%

merge_return_later (_Key, _EarlierValue, LaterValue) ->
  LaterValue.

short_circuit_always (_Key, _Value) ->
  true.

short_circuit_never (_Key, _Value) ->
  false.

delete_never (_Key, _Value) ->
  false.

delete_empty (_Key, <<>>) ->
  true;
delete_empty (_Key, _Value) ->
  false.

merge_uint64_vector_sum (_Key, B1, B2) ->
  uint64_vector_sum (B1, B2, []).

uint64_vector_sum (<< C1:64/big-unsigned-integer, B1/binary >>,
		   << C2:64/big-unsigned-integer, B2/binary >>,
		   Acc) ->
  Sum = C1 + C2,
  uint64_vector_sum (B1, B2, [ << Sum:64/big-unsigned-integer >> | Acc ]);
uint64_vector_sum (B1, <<>>, Acc) when B1 =/= <<>> ->
  iolist_to_binary (lists:reverse ([ B1 | Acc ]));
uint64_vector_sum (<<>>, B2, Acc) when B2 =/= <<>> ->
  iolist_to_binary (lists:reverse ([ B2 | Acc ]));
uint64_vector_sum (<<>>, <<>>, Acc) ->
  iolist_to_binary (lists:reverse (Acc)).

%
% tests
%

-ifdef (EUNIT).

basic_test () ->
  ok.

-endif.
